package com.hwtx.protocol.dubbo.code;

import com.hzgj.bcl.soa.net.Decoder;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;

import static com.hwtx.protocol.dubbo.code.DubboConstants.DUBBO_PROTOCOL;
import static com.hwtx.protocol.dubbo.code.DubboConstants.FLAG_EVENT;
import static com.hwtx.protocol.dubbo.code.DubboConstants.FLAG_REQUEST;
import static com.hwtx.protocol.dubbo.code.DubboConstants.HEADER_LENGTH;
import static com.hwtx.protocol.dubbo.code.DubboConstants.MAGIC;
import static com.hwtx.protocol.dubbo.code.DubboConstants.RESPONSE_OK;
import static com.hwtx.protocol.dubbo.code.DubboConstants.SERIALIZATION_MASK;
import static com.hzgj.bcl.soa.net.DefaultEncoder.checkPayload;
import static com.hzgj.bcl.soa.util.Constants.MESSAGE_HEADER_PROTOCOL;

public class DubboDecoder extends Decoder {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private State state = State.Header;
    private int dataLen;
    private byte serializationId;
    private byte event = 0;
    private int message_type;
    private long sid;

    @Override
    public String getName() {
        return DUBBO_PROTOCOL;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        switch (state) {
            case Header:
                if (in.readableBytes() < HEADER_LENGTH) {
                    return;
                }
                short magic = in.getShort(in.readerIndex());
                if (magic != MAGIC) {
                    logger.error("data is not according to dubbo handler.");
                    ctx.channel().close();
                    return;
                }
                in.readShort();//跳过magic
                byte header2Flag = in.readByte();
                serializationId = (byte) (header2Flag & SERIALIZATION_MASK);
                boolean isRequest = ((header2Flag & FLAG_REQUEST) == FLAG_REQUEST);
                if ((header2Flag & FLAG_EVENT) != 0) {
                    event = 1;
                }
                if (isRequest) {
                    //status
                    in.readByte();
                    message_type = Message.REQ;
                } else {
                    byte status = in.readByte();
                    if (status == RESPONSE_OK) {
                        message_type = Message.RSP;
                    } else {
                        message_type = Message.EXC_RSP;
                    }
                }
                sid = in.readLong();
                dataLen = in.readInt();
                checkPayload(ctx.channel(), dataLen);
                state = State.Body;
                break;
            case Body:
                if (in.readableBytes() < dataLen) {
                    return;
                }
                byte[] body = new byte[dataLen];
                in.readBytes(body);
                Message message = new Message();
                message.putHeader(Constants.MESSAGE_HEADER_SERIALIZATION_ID, serializationId)
                        .putHeader(Constants.MESSAGE_HEADER_EVENT, event)
                        .putHeader(Constants.MESSAGE_TYPE, message_type)
                        .putHeader(MESSAGE_HEADER_PROTOCOL, DUBBO_PROTOCOL)
                        .putHeader(Constants.MESSAGE_SID, sid).setPayload(body);

                out.add(message);
                state = State.Header;
                break;
        }
    }

    private enum State {
        Header,
        Body
    }
}